package com.duobaoyu.dby.supply.biz.service.queue.core;


import com.duobaoyu.dby.supply.core.delay.annotation.RedissonListener;
import com.duobaoyu.dby.supply.core.delay.message.RedissonMessage;
import com.duobaoyu.dby.supply.db.entity.message.DelayMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;

import static com.duobaoyu.dby.supply.biz.service.queue.core.DistributionQueueConstant.DISTRIBUTION_QUEUE_MESSAGE_ID;

/**
 * Created by shuming on 2023/9/27 15:56
 */
@Slf4j
@RequiredArgsConstructor
public class DistributionQueueJobConsumerProxy {

    private final DistributionQueueJobConsumerWork work;

    @RedissonListener(queues = "${distribution.queue.listener}")
    public void handle(RedissonMessage redissonMessage) {
        Object msgId = null;
        try {
            msgId = redissonMessage.getHeaders().get(DISTRIBUTION_QUEUE_MESSAGE_ID);
            DelayMessage delayMessage = work.loadJobMessage((String) msgId);
            if (Objects.isNull(delayMessage)) {
                log.warn("数据异常,msgId={}", msgId);
                return;
            }
            work.consume(delayMessage);
        } catch (Exception e) {
            log.warn("consume distribution queue throw exception,msgId={}", msgId, e);
            throw e;
        }
    }
}
